Reviewing files that changed from the base of the PR and between ea8a9ca and 02ac489.

• src/main/resources/application.conf (1 hunks)
• src/main/scala/dpla/api/v2/search/ElasticSearchClient.scala (8 hunks)

• GitHub Check: Agent

Reviewing files that changed from the base of the PR and between 02ac489 and 93ec885.

• src/main/scala/dpla/api/v2/search/ElasticSearchClient.scala (8 hunks)

src/main/scala/dpla/api/v2/search/ElasticSearchClient.scala (5)

12-13: LGTM!

The imports are appropriate for the semaphore-based concurrency limiting feature.

---

20-35: Past review concerns addressed.

The environment variable parsing now includes proper error handling for invalid numeric values and validation for positive integers. The implementation correctly addresses the concerns raised in previous reviews.

---

44-66: Past review concerns fully addressed.

The withConcurrencyLimit implementation now correctly handles all previously flagged issues:

• Uses tryAcquire with timeout instead of blocking acquire(), preventing actor thread blocking
• Includes try-catch to ensure permit release even if Future construction fails
• Provides clear error messaging when concurrency limit is exceeded

The implementation is robust and addresses concurrency concerns appropriately.

---

133-133: LGTM!

The ExecutionContext is correctly added as an implicit parameter (required by withConcurrencyLimit), and the ES HTTP request is properly wrapped with the concurrency limiter.

Also applies to: 142-144

---

188-188: LGTM!

The concurrency limiter is consistently applied across all ES request paths (fetch, multi-fetch, and random), with the necessary ExecutionContext implicit parameter added to each method. The implementation maintains consistency with the search request pattern.

Also applies to: 192-194, 240-240, 249-251, 294-294, 303-305

Reviewing files that changed from the base of the PR and between 93ec885 and 939ef26.

• src/main/scala/dpla/api/v2/search/ElasticSearchClient.scala (9 hunks)

src/main/scala/dpla/api/v2/search/ElasticSearchClient.scala (7)

25-26: LGTM: Imports support concurrency changes.

The ExecutionContext import is necessary for the implicit ec parameters added to each handler method, and the LoggerFactory import supports the error logging added for configuration validation.

---

32-32: LGTM: Logger properly initialized.

The SLF4J logger is correctly initialized and used appropriately for configuration validation errors.

---

40-61: LGTM: Configuration parsing is robust.

The environment variable parsing correctly handles invalid inputs with safe fallbacks. The validation ensures positive values and provides clear error messages when falling back to defaults.

---

190-192: Verify ConcurrencyLimiter API supports this usage pattern.

The code uses concurrencyLimiter { ... } syntax, which requires ConcurrencyLimiter to either:

1. Have an apply method: def apply[T](f: => Future[T]): Future[T]
2. Be assigned to a function value rather than a class instance

When implementing the missing ConcurrencyLimiter class, ensure its API supports this by-name parameter pattern. Based on past reviews, this should wrap the HTTP request with semaphore-based concurrency control.

The implementation should handle:

• Acquiring permits with timeout (using semaphoreTimeoutSeconds)
• Releasing permits on both success and failure
• Avoiding permit leaks if Future construction fails
• Non-blocking operation to avoid starving Akka threads

---

181-181: LGTM: ExecutionContext properly added to all handlers.

Each handler method correctly adds an implicit ExecutionContext derived from the actor system. This is necessary for Future operations and follows Akka best practices by using system.executionContext.

Also applies to: 235-235, 285-285, 338-338

---

166-169: LGTM: Documentation improvements.

The added Scaladoc comments correctly describe the per-session actor pattern, clarifying that each session has its own internal state and ActorRef.

Also applies to: 271-274

---

90-92: Ensure ConcurrencyLimiter addresses past review concerns.

Previous reviews raised several critical concerns about the concurrency limiting implementation (semaphore blocking, permit leaks, thread starvation). Those details are no longer visible in this file, suggesting they were refactored into the missing ConcurrencyLimiter class.

When providing the ConcurrencyLimiter implementation, ensure it addresses:

1. Non-blocking operation: Avoid acquire() which blocks Akka threads; use tryAcquire() with timeout
2. Permit leak prevention: Ensure permits are released even if Future construction fails, not just in andThen callback
3. Thread safety: Properly handle concurrent access to the semaphore
4. Timeout handling: Fail fast when semaphoreTimeoutSeconds is exceeded rather than blocking indefinitely
5. Akka integration: Consider using Akka's native backpressure mechanisms or a dedicated blocking dispatcher

Past reviews suggested patterns like:

def apply[T](f: => Future[T])(implicit ec: ExecutionContext): Future[T] = {
  if (!semaphore.tryAcquire(timeoutSeconds, TimeUnit.SECONDS)) {
    Future.failed(new TimeoutException("ES concurrency limit exceeded"))
  } else {
    try {
      val future = f
      future.onComplete(_ => semaphore.release())(ec)
      future
    } catch {
      case e: Throwable =>
        semaphore.release()
        Future.failed(e)
    }
  }
}